package com.alibaba.datax.plugin.writer.dmwriter;

import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.ListUtil;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
import com.alibaba.datax.plugin.rdbms.util.RdbmsException;
import com.alibaba.datax.plugin.rdbms.writer.Constant;
import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Triple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 达梦数据库Writer插件
 */
public class DmWriter extends Writer {
    private static final Logger LOG = LoggerFactory.getLogger(DmWriter.class);
    private static Connection CONNECTION = null;

    private static Connection getConnection(String jdbcUrl, String username, String password) {
        try {
            if (CONNECTION == null || CONNECTION.isClosed()) {
                DbUtils.loadDriver(Key.DM_JDBC_DRIVER);
//                DbUtils.loadDriver("dm.jdbc.driver.DmDriver");
            }
            CONNECTION = DriverManager.getConnection(jdbcUrl, username, password);
            LOG.info("成功连接到达梦数据库: {}", jdbcUrl);
        } catch (SQLException e) {
            throw DataXException.asDataXException(DmWriterErrorCode.CONNECTION_ERROR, "创建数据库连接失败: " + e.getMessage(), e);
        }
        return CONNECTION;

    }

    public static class Job extends Writer.Job {
        private Configuration originalConfig;
        protected String username;
        protected String password;
        private DmMetadataUtil metadataUtil;

        @Override
        public void init() {
            this.originalConfig = super.getPluginJobConf();
            LOG.info("DmWriter.Job init with config: {}", originalConfig);
            // 检查 username/password 配置（必填）
            originalConfig.getNecessaryValue(Key.USERNAME, DBUtilErrorCode.REQUIRED_VALUE);
            originalConfig.getNecessaryValue(Key.PASSWORD, DBUtilErrorCode.REQUIRED_VALUE);
            username = originalConfig.getString(Key.USERNAME);
            password = originalConfig.getString(Key.PASSWORD);
            this.metadataUtil = new DmMetadataUtil(originalConfig);
            doCheckBatchSize();

            simplifyConf();

            dealColumnConf();

//            dealWriteMode();
        }

        public void doCheckBatchSize() {
            // 检查batchSize 配置（选填，如果未填写，则设置为默认值）
            int batchSize = originalConfig.getInt(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE);
            if (batchSize < 1) {
                throw DataXException.asDataXException(DBUtilErrorCode.ILLEGAL_VALUE, String.format(
                        "您的batchSize配置有误. 您所配置的写入数据库表的 batchSize:%s 不能小于1. 推荐配置范围为：[100-1000], 该值越大, 内存溢出可能性越大. 请检查您的配置并作出修改.", batchSize));
            }

            originalConfig.set(Key.BATCH_SIZE, batchSize);
        }

        private void simplifyConf() {
            List<Object> connections = originalConfig.getList(Constant.CONN_MARK, Object.class);
            int tableNum = 0;
            // 获取所有表的主键信息
            Map<String, List<String>> allPrimaryKeys = new HashMap<>();
            for (int i = 0; i < connections.size(); i++) {
                Configuration connConf = Configuration.from(connections.get(i).toString());

                String jdbcUrl = connConf.getString(Key.JDBC_URL);

                if (StringUtils.isBlank(jdbcUrl)) {
                    throw DataXException.asDataXException(DBUtilErrorCode.REQUIRED_VALUE, "您未配置的写入数据库表的 jdbcUrl.");
                }
                List<String> tables = connConf.getList(Key.TABLE, String.class);
                if (null == tables || tables.isEmpty()) {
                    throw DataXException.asDataXException(DBUtilErrorCode.REQUIRED_VALUE, "您未配置写入数据库表的表名称. 根据配置DataX找不到您配置的表. 请检查您的配置并作出修改.");
                }
                // 解析表参数是否含有模式名，如果没有，则增加
                /*List<String> expandedTables = addDMSchema(tables, jdbcUrl);
                tableNum += expandedTables.size();
                originalConfig.set(String.format("%s[%d].%s", Constant.CONN_MARK, i, Key.TABLE), expandedTables);*/
                tableNum += tables.size();
                originalConfig.set(String.format("%s[%d].%s", Constant.CONN_MARK, i, Key.TABLE), tables);
                // 获取主键
                Map<String, List<String>> primaryKeys = metadataUtil.getBatchPrimaryKeys(tables);
                for (Map.Entry<String, List<String>> entry : primaryKeys.entrySet()) {
                    allPrimaryKeys.putIfAbsent(entry.getKey(), entry.getValue());
                }
            }
            originalConfig.set(Constant.TABLE_NUMBER_MARK, tableNum);
            originalConfig.set(Key.PRIMARY_KEYS, allPrimaryKeys);
        }

        private void dealColumnConf() {
            String jdbcUrl = originalConfig.getString(String.format("%s[0].%s", Constant.CONN_MARK, Key.JDBC_URL));
            String username = originalConfig.getString(Key.USERNAME);
            String password = originalConfig.getString(Key.PASSWORD);
            String oneTable = originalConfig.getString(String.format("%s[0].%s[0]", Constant.CONN_MARK, Key.TABLE));


            Connection conn = null;
            try {
                conn = getConnection(jdbcUrl, username, password);

                List<String> userConfiguredColumns = originalConfig.getList(Key.COLUMN, String.class);
                if (null == userConfiguredColumns || userConfiguredColumns.isEmpty()) {
                    throw DataXException.asDataXException(DBUtilErrorCode.ILLEGAL_VALUE, "您的配置文件中的列配置信息有误. 因为您未配置写入数据库表的列名称，DataX获取不到列信息. 请检查您的配置并作出修改.");
                } else {
                    boolean isPreCheck = originalConfig.getBool(Key.DRYRUN, false);
                    List<String> allColumns = getTableColumnsByConn(conn, oneTable);
                    LOG.info("table:[{}] all columns:[\n{}\n].", oneTable, StringUtils.join(allColumns, ","));

                    if (1 == userConfiguredColumns.size() && "*".equals(userConfiguredColumns.get(0))) {
                        LOG.warn("您的配置文件中的列配置信息存在风险. 因为您配置的写入数据库表的列为*，当您的表字段个数、类型有变动时，可能影响任务正确性甚至会运行出错。请检查您的配置并作出修改.");

                        // 回填其值，需要以 String 的方式转交后续处理
                        originalConfig.set(Key.COLUMN, allColumns);
                    } else if (userConfiguredColumns.size() > allColumns.size()) {
                        throw DataXException.asDataXException(DBUtilErrorCode.ILLEGAL_VALUE,
                                String.format("您的配置文件中的列配置信息有误. 因为您所配置的写入数据库表的字段个数:%s 大于目的表的总字段总个数:%s. 请检查您的配置并作出修改.",
                                        userConfiguredColumns.size(), allColumns.size()));
                    } else {
                        // 确保用户配置的 column 不重复
                        ListUtil.makeSureNoValueDuplicate(userConfiguredColumns, false);
                        try {
                            // 检查列是否都为数据库表中正确的列（通过执行一次 select column from table 进行判断）
                            DBUtil.getColumnMetaData(conn, oneTable, StringUtils.join(userConfiguredColumns, ","));

                        } finally {
                            DbUtils.closeQuietly(conn);
                        }

                    }
                }


            } catch (Exception e) {
                throw DataXException.asDataXException(DmWriterErrorCode.CONNECTION_ERROR, "连接达梦数据库失败: " + e.getMessage(), e);
            } finally {
                DbUtils.closeQuietly(conn);
            }


        }

        private void dealWriteMode() {
            List<String> columns = originalConfig.getList(Key.COLUMN, String.class);

            String jdbcUrl = originalConfig.getString(String.format("%s[0].%s", Constant.CONN_MARK, Key.JDBC_URL));

            // 默认为：insert 方式
            String writeMode = originalConfig.getString(Key.WRITE_MODE, "INSERT");

            List<String> valueHolders = new ArrayList<String>(columns.size());
            for (int i = 0; i < columns.size(); i++) {
                valueHolders.add("?");
            }

            boolean forceUseUpdate = false;

            String writeDataSqlTemplate = getWriteTemplate(columns, valueHolders, writeMode, forceUseUpdate);

            LOG.info("Write data [\n{}\n], which jdbcUrl like:[{}]", writeDataSqlTemplate, jdbcUrl);

            originalConfig.set(Constant.INSERT_OR_REPLACE_TEMPLATE_MARK, writeDataSqlTemplate);
        }

        // 获取表主键，如果不存在，则报错 todo
        public String getWriteTemplate(List<String> columnHolders, List<String> valueHolders, String writeMode, boolean forceUseUpdate) {
            boolean isWriteModeLegal = writeMode.trim().toLowerCase().startsWith("insert") || writeMode.trim().toLowerCase().startsWith("replace") || writeMode.trim().toLowerCase().startsWith("update");

            if (!isWriteModeLegal) {
                throw DataXException.asDataXException(DBUtilErrorCode.ILLEGAL_VALUE, String.format("您所配置的 writeMode:%s 错误. 因为DataX 目前仅支持replace,update 或 insert 方式. 请检查您的配置并作出修改.", writeMode));
            }
            // && writeMode.trim().toLowerCase().startsWith("replace")
            String writeDataSqlTemplate = null;

            if (Key.WRITE_MODE_REPLACE.equals(writeMode) || Key.WRITE_MODE_UPDATE.equals(writeMode)) {
//                Map<String, List<String>> map = originalConfig.getMap(Key.PRIMARY_KEYS, new HashMap<String, List<String>>(), List.class);
                // 达梦使用MERGE INTO实现REPLACE

                writeDataSqlTemplate = new StringBuilder()
                        .append("MERGE INTO  %s USING DUAL ON (id = ? ) WHEN MATCHED THEN UPDATE SET ")
                        .append(columnHolders.stream().map(s -> s + " = ?").collect(Collectors.joining(",")))
                        .append(" WHEN NOT MATCHED THEN INSERT (")
                        .append(StringUtils.join(columnHolders, ","))

                        .append(") VALUES (")
                        .append(StringUtils.join(valueHolders, ","))
                        .append(" )")
                        .toString();
            } else {
                // 默认使用INSERT
                writeDataSqlTemplate = new StringBuilder()
                        .append("INSERT INTO %s (").append(StringUtils.join(columnHolders, ","))
                        .append(") VALUES(").append(StringUtils.join(valueHolders, ","))
                        .append(")")
                        .toString();
            }
            return writeDataSqlTemplate;
        }

        public List<String> getTableColumnsByConn(Connection conn, String tableName) {
            List<String> columns = new ArrayList<String>();
            Statement statement = null;
            ResultSet rs = null;
            String queryColumnSql = null;
            try {
                statement = conn.createStatement();
                queryColumnSql = String.format("select * from %s where 1=2", tableName);
                rs = statement.executeQuery(queryColumnSql);
                ResultSetMetaData rsMetaData = rs.getMetaData();
                for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
                    columns.add(rsMetaData.getColumnName(i + 1));
                }

            } catch (SQLException e) {
                throw DataXException.asDataXException(DBUtilErrorCode.SQL_EXECUTE_FAIL, "执行的SQL为: " + queryColumnSql + " 具体错误信息为：" + e, e);
            } finally {
                DbUtils.closeQuietly(rs);
                DbUtils.closeQuietly(statement);
            }

            return columns;
        }

        @Override
        public void prepare() {
            LOG.info("DmWriter.Job prepare");

            // 执行前置SQL
            int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
            if (tableNumber == 1) {
                List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
                Configuration connConf = Configuration.from(conns.get(0).toString());

                // 这里的 jdbcUrl 已经 append 了合适后缀参数
                String jdbcUrl = connConf.getString(Key.JDBC_URL);
                originalConfig.set(Key.JDBC_URL, jdbcUrl);

                String table = connConf.getList(Key.TABLE, String.class).get(0);
                originalConfig.set(Key.TABLE, table);

                List<String> preSqls = originalConfig.getList(Key.PRE_SQL, String.class);
                List<String> renderedPreSqls = WriterUtil.renderPreOrPostSqls(preSqls, table);

                originalConfig.remove(Constant.CONN_MARK);
                if (null != renderedPreSqls && !renderedPreSqls.isEmpty()) {
                    // 说明有 preSql 配置，则此处删除掉
                    originalConfig.remove(Key.PRE_SQL);
                    Connection conn = null;
                    Statement stmt = null;
                    try {
                        conn = getConnection(jdbcUrl, username, password);
                        LOG.info("Begin to execute preSqls:[{}]. context info:{}.", StringUtils.join(renderedPreSqls, ";"), jdbcUrl);
                        stmt = conn.createStatement();
                        for (String preSql : preSqls) {
                            stmt.execute(preSql);
                        }
                    } catch (Exception e) {
                        LOG.warn("执行前置SQL失败: {}", e.getMessage());
                    } finally {
                        DbUtils.closeQuietly(stmt);
                        DbUtils.closeQuietly(conn);
                    }
                }
            }

            LOG.debug("After job prepare(), originalConfig now is:[\n{}\n]", originalConfig.toJSON());
        }

        @Override
        public List<Configuration> split(int mandatoryNumber) {
            LOG.info("DmWriter.Job split into {} tasks", mandatoryNumber);
            List<Configuration> splitResultConfigs = new ArrayList<Configuration>();

            int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);

            //处理单表的情况
            if (tableNumber == 1) {
                //由于在之前的  master prepare 中已经把 table,jdbcUrl 提取出来，所以这里处理十分简单
                for (int j = 0; j < mandatoryNumber; j++) {
                    splitResultConfigs.add(originalConfig.clone());
                }

                return splitResultConfigs;
            }

            if (tableNumber != mandatoryNumber) {
                throw DataXException.asDataXException(DBUtilErrorCode.CONF_ERROR,
                        String.format("您的配置文件中的列配置信息有误. 您要写入的目的端的表个数是:%s , 但是根据系统建议需要切分的份数是：%s. 请检查您的配置并作出修改.",
                                tableNumber, mandatoryNumber));
            }

            String jdbcUrl;
            List<String> preSqls = originalConfig.getList(Key.PRE_SQL, String.class);
            List<String> postSqls = originalConfig.getList(Key.POST_SQL, String.class);

            List<Object> conns = originalConfig.getList(Constant.CONN_MARK,
                    Object.class);

            for (Object conn : conns) {
                Configuration sliceConfig = originalConfig.clone();

                Configuration connConf = Configuration.from(conn.toString());
                jdbcUrl = connConf.getString(Key.JDBC_URL);
                sliceConfig.set(Key.JDBC_URL, jdbcUrl);

                sliceConfig.remove(Constant.CONN_MARK);

                List<String> tables = connConf.getList(Key.TABLE, String.class);

                for (String table : tables) {
                    Configuration tempSlice = sliceConfig.clone();
                    tempSlice.set(Key.TABLE, table);
                    tempSlice.set(Key.PRE_SQL, renderPreOrPostSqls(preSqls, table));
                    tempSlice.set(Key.POST_SQL, renderPreOrPostSqls(postSqls, table));

                    splitResultConfigs.add(tempSlice);
                }

            }

            return splitResultConfigs;
            /*List<Configuration> configurations = new ArrayList<Configuration>();

            for (int i = 0; i < mandatoryNumber; i++) {
                configurations.add(originalConfig.clone());
            }

            return configurations;*/
        }

        public List<String> renderPreOrPostSqls(List<String> preOrPostSqls, String tableName) {
            if (null == preOrPostSqls) {
                return Collections.emptyList();
            }

            List<String> renderedSqls = new ArrayList<String>();
            for (String sql : preOrPostSqls) {
                //preSql为空时，不加入执行队列
                if (StringUtils.isNotBlank(sql)) {
                    renderedSqls.add(sql.replace(Constant.TABLE_NAME_PLACEHOLDER, tableName));
                }
            }

            return renderedSqls;
        }

        @Override
        public void post() {
            LOG.info("DmWriter.Job post");
            int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
            if (tableNumber == 1) {

                // 已经由 prepare 进行了appendJDBCSuffix处理
                String jdbcUrl = originalConfig.getString(Key.JDBC_URL);

                String table = originalConfig.getString(Key.TABLE);

                List<String> postSqls = originalConfig.getList(Key.POST_SQL, String.class);
                List<String> renderedPostSqls = WriterUtil.renderPreOrPostSqls(postSqls, table);

                if (null != renderedPostSqls && !renderedPostSqls.isEmpty()) {
                    // 说明有 postSql 配置，则此处删除掉
                    originalConfig.remove(Key.POST_SQL);

                    Connection conn = getConnection(jdbcUrl, username, password);

                    LOG.info("Begin to execute postSqls:[{}]. context info:{}.", StringUtils.join(renderedPostSqls, ";"), jdbcUrl);

                    Statement stmt = null;
                    String currentSql = null;
                    try {
                        stmt = conn.createStatement();
                        for (String sql : postSqls) {
                            currentSql = sql;
                            stmt.execute(sql);
                        }
                    } catch (Exception e) {
                        throw DataXException.asDataXException(DBUtilErrorCode.SQL_EXECUTE_FAIL, "执行的SQL为: " + currentSql + " 具体错误信息为：" + e, e);
                    } finally {
                        DbUtils.closeQuietly(stmt);
                    }
                    DbUtils.closeQuietly(conn);
                }
            }
        }

        @Override
        public void destroy() {
            LOG.info("DmWriter.Job destroy");
        }
    }

    public static class Task extends Writer.Task {
        private Configuration taskConfig;
        private Connection connection;
        private PreparedStatement preparedStatement;
        private String tableName;
        private List<String> columns;
        private String writeMode;
        private int batchSize;
        private int batchCount = 0;


        private static final String VALUE_HOLDER = "?";

        protected String username;
        protected String password;
        protected String jdbcUrl;
        protected String table;
        protected List<String> preSqls;
        protected List<String> postSqls;
        protected int batchByteSize;
        protected int columnNumber = 0;
        protected TaskPluginCollector taskPluginCollector;

        // 作为日志显示信息时，需要附带的通用信息。比如信息所对应的数据库连接等信息，针对哪个表做的操作
        protected static String BASIC_MESSAGE;

        protected static String INSERT_OR_REPLACE_TEMPLATE;

        protected String writeRecordSql;
        protected boolean emptyAsNull;
        protected Triple<List<String>, List<Integer>, List<String>> resultSetMetaData;

        private static final Logger TASK_LOG = LoggerFactory.getLogger(Task.class);

        @Override
        public void init() {
            this.taskConfig = super.getPluginJobConf();
            TASK_LOG.info("DmWriter.Task init with config: {}", taskConfig);

            // 初始化配置 初始化数据库连接
            initConfig();

            // 准备SQL语句
            prepareStatement();
        }

        private void initConfig() {
            this.username = taskConfig.getString(Key.USERNAME);
            this.password = taskConfig.getString(Key.PASSWORD);
            this.jdbcUrl = taskConfig.getString(Key.JDBC_URL);
            this.tableName = taskConfig.getString(Key.TABLE);
            this.columns = taskConfig.getList(Key.COLUMN, String.class);
            this.writeMode = taskConfig.getString(Key.WRITE_MODE, Key.DEFAULT_WRITE_MODE);
            this.batchSize = taskConfig.getInt(Key.BATCH_SIZE, Key.DEFAULT_BATCH_SIZE);

            if (columns == null || columns.isEmpty()) {
                throw DataXException.asDataXException(DmWriterErrorCode.BAD_SQL_SYNTAX, "column 配置不能为空");
            }

            this.table = taskConfig.getString(Key.TABLE);

            this.columns = taskConfig.getList(Key.COLUMN, String.class);
            this.columnNumber = this.columns.size();

            this.preSqls = taskConfig.getList(Key.PRE_SQL, String.class);
            this.postSqls = taskConfig.getList(Key.POST_SQL, String.class);
            this.batchSize = taskConfig.getInt(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE);
            this.batchByteSize = taskConfig.getInt(Key.BATCH_BYTE_SIZE, Constant.DEFAULT_BATCH_BYTE_SIZE);

            writeMode = taskConfig.getString(Key.WRITE_MODE, "INSERT");
            emptyAsNull = taskConfig.getBool(Key.EMPTY_AS_NULL, true);
            INSERT_OR_REPLACE_TEMPLATE = taskConfig.getString(Constant.INSERT_OR_REPLACE_TEMPLATE_MARK);
            if (StringUtils.isNotBlank(INSERT_OR_REPLACE_TEMPLATE)) {
                this.writeRecordSql = String.format(INSERT_OR_REPLACE_TEMPLATE, this.table);
            }

            this.connection = getConnection(jdbcUrl, username, password);
            BASIC_MESSAGE = String.format("jdbcUrl:[%s], table:[%s]", this.jdbcUrl, this.table);

        }

        /*
         * "MERGE INTO user_info target USING (SELECT ? as id, ? as name, ? as age, ? as email FROM dual) source ON (target.id = source.id)
         * WHEN MATCHED THEN UPDATE SET target.name=source.name, target.age=source.age, target.email=source.email
         * WHEN NOT MATCHED THEN INSERT (id, name, age, email) VALUES (source.id, source.name, source.age, source.email)"
         *
         *
         * */
        private void prepareStatement() {
            try {
                String sql = null;
                if (Key.WRITE_MODE_REPLACE.equals(writeMode) || Key.WRITE_MODE_UPDATE.equals(writeMode)) {
                    Map<String, Object> map = taskConfig.getMap(Key.PRIMARY_KEYS);
                    if (map == null) {
                        throw new RuntimeException("");

                    }
                    Object o = map.get(table);
                    if (o == null) {
                        throw new RuntimeException("");
                    }
                    List<String> primaryKey = JSON.parseArray(o.toString(), String.class);

                    sql = new StringBuilder()
                            .append("MERGE INTO ").append(tableName).append(" target ")
                            .append(" USING ( SELECT ")
                            .append(columns.stream().map(s -> " ? as " + s).collect(Collectors.joining(",")))

                            .append(" FROM dual ) source on (")
                            // 条件
                            .append(primaryKey.stream().map(s -> " target." + s + " = source." + s).collect(Collectors.joining(",")))

                            .append(" )")

                            .append("WHEN MATCHED THEN UPDATE SET ")
                            // 更新字段值
                            .append(columns.stream().filter(s -> !primaryKey.contains(s)).map(s -> " target." + s + " = source." + s).collect(Collectors.joining(",")))
                            // 插入值
                            .append(" WHEN NOT MATCHED THEN INSERT (")
                            .append(StringUtils.join(columns, ","))
                            .append(") VALUES (")

                            .append(columns.stream().map(s -> "source." + s).collect(Collectors.joining(",")))

                            .append(" )")
                            .toString();
                } else {// 其他都是插入
                    sql = new StringBuilder()
                            .append("INSERT INTO ").append(tableName).append(" (").append(StringUtils.join(columns, ","))
                            .append(") VALUES(").append(columns.stream().map(s -> "?").collect(Collectors.joining(",")))
                            .append(")")
                            .toString();
                }


                TASK_LOG.info("准备SQL语句: {}", sql);
                this.preparedStatement = connection.prepareStatement(sql);
            } catch (SQLException e) {
                throw DataXException.asDataXException(DmWriterErrorCode.BAD_SQL_SYNTAX, "准备SQL语句失败: " + e.getMessage(), e);
            }
        }

        @Override
        public void prepare() {
            TASK_LOG.info("DmWriter.Task Begin to execute preSqls:[{}]. context info:{}.", StringUtils.join(this.preSqls, ";"), BASIC_MESSAGE);
            if (preSqls == null || preSqls.isEmpty()) {
                return;
            }
            Connection connection = getConnection(jdbcUrl, username, password);
            try {
                Statement stmt = connection.createStatement();
                List<String> renderedPreSqls = WriterUtil.renderPreOrPostSqls(preSqls, table);
                String querySql = null;
                for (String preSql : renderedPreSqls) {
                    stmt.execute(preSql);
                }
                int tableNumber = taskConfig.getInt(Constant.TABLE_NUMBER_MARK);
                if (tableNumber != 1) {

                    throw DataXException.asDataXException(DmWriterErrorCode.BAD_SQL_SYNTAX, "执行的SQL为: " + querySql);
                }
            } catch (Exception e) {
                throw DataXException.asDataXException(DmWriterErrorCode.BAD_SQL_SYNTAX, " 具体错误信息为：" + e, e);
            } finally {
                DbUtils.closeQuietly(connection);
            }


        }

        @Override
        public void startWrite(RecordReceiver recordReceiver) {
            TASK_LOG.info("DmWriter.Task start writing data");

            Record record;
            try {
                while ((record = recordReceiver.getFromReader()) != null) {
                    writeRecord(record);
                }
                // 提交最后一批数据
                flush();
            } catch (Exception e) {
                throw DataXException.asDataXException(DmWriterErrorCode.DATA_ERROR, "写入数据失败: " + e.getMessage(), e);
            }
        }

        private void writeRecord(Record record) {
            try {
                // 设置PreparedStatement参数
                for (int i = 0; i < columns.size(); i++) {
                    setPreparedStatement(i + 1, record.getColumn(i));
                }

                preparedStatement.addBatch();
                batchCount++;

                if (batchCount >= batchSize) {
                    flush();
                }
            } catch (Exception e) {
                super.getTaskPluginCollector().collectDirtyRecord(record, e);
                TASK_LOG.warn("记录写入失败: {}", e.getMessage());
            }
        }

        private void setPreparedStatement(int index, com.alibaba.datax.common.element.Column column)
                throws SQLException {
            if (column == null || column.getRawData() == null) {
                preparedStatement.setNull(index, Types.NULL);
                return;
            }
            com.alibaba.datax.common.element.Column.Type type = column.getType();
            switch (type) {
                case LONG:
                    preparedStatement.setLong(index, column.asLong());
                    break;
                case DOUBLE:
                    preparedStatement.setDouble(index, column.asDouble());
                    break;
                case STRING:
                    preparedStatement.setString(index, column.asString());
                    break;
                case DATE:
                    preparedStatement.setTimestamp(index, new Timestamp(column.asDate().getTime()));
                    break;
                case BOOL:
                    preparedStatement.setBoolean(index, column.asBoolean());
                    break;
                case BYTES:
                    preparedStatement.setBytes(index, column.asBytes());
                    break;
                default:
                    preparedStatement.setString(index, column.asString());
            }
        }

        private void flush() {
            if (batchCount == 0) {
                return;
            }

            try {
                int[] results = preparedStatement.executeBatch();
                connection.commit();
                TASK_LOG.info("成功提交 {} 条数据", batchCount);
                batchCount = 0;
            } catch (SQLException e) {
                try {
                    connection.rollback();
                } catch (SQLException rollbackEx) {
                    TASK_LOG.error("回滚事务失败: {}", rollbackEx.getMessage());
                }
                throw DataXException.asDataXException(DmWriterErrorCode.BATCH_INSERT_DATA_ERROR, "批量提交数据失败: " + e.getMessage(), e);
            }
        }

        @Override
        public void post() {
            TASK_LOG.info("DmWriter.Task post");
            // 执行后置SQL
            executePostSql();
        }

        private void executePostSql() {
            List<String> postSqls = taskConfig.getList(Key.POST_SQL, String.class);
            if (postSqls == null || postSqls.isEmpty()) {
                return;
            }

            Statement stmt = null;
            try {
                stmt = connection.createStatement();
                for (String sql : postSqls) {
                    if (sql != null && !sql.trim().isEmpty()) {
                        TASK_LOG.info("执行后置SQL: {}", sql);
                        stmt.execute(sql);
                    }
                }
                connection.commit();
            } catch (SQLException e) {
                TASK_LOG.warn("执行后置SQL失败: {}", e.getMessage());
            } finally {
                DbUtils.closeQuietly(stmt);
            }
        }

        @Override
        public void destroy() {
            TASK_LOG.info("DmWriter.Task destroy");
            // 清理资源
            DbUtils.closeQuietly(preparedStatement);
            DbUtils.closeQuietly(connection);
        }
    }
}